package com.xl.flinkdemo.source;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;

import java.util.Properties;

/**
 * @ClassName SourceTest_Collecttion2
 * @Description TODO
 * @Author hxl
 * @Date 2021/4/14 15:08
 * Version 1.0
 **/
public class SourceTest_Collecttion_kafka {

  public static void main(String[] args) throws Exception {
    //创建执行环境
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    //按顺序输出
    env.setParallelism(1);
    Properties properties = new Properties();
    properties.setProperty("boot","");

    //从kafka中读取数据
    DataStream dataStream = env.addSource(new FlinkKafkaConsumer011<String>("sensor",new SimpleStringSchema(),properties));

    //打印输出
    dataStream.print();

    //执行
    env.execute("jobName_TestSourceJob");
  }

}
